/**
 * 
 */
package org.youisoft.summer.evaluator.function.control;

import java.security.PrivilegedAction;
import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;
import org.youisoft.summer.evaluator.InvokeException;
import org.youisoft.summer.evaluator.context.CalContext;
import org.youisoft.summer.evaluator.function.AbstractFunction;
import org.youisoft.summer.evaluator.operator.Operator;
import org.youisoft.summer.evaluator.operator.base.ClosureOperator;
import org.youisoft.summer.evaluator.operator.base.MapOperator;
import org.youisoft.summer.evaluator.operator.base.NumberOperator;
import org.youisoft.summer.evaluator.operator.base.StringOperator;

/**
 * @author zijiang.jl
 *
 */
public class PThreadFunction extends AbstractFunction{
	public static final StringOperator POOL=new StringOperator("pool");
	public static final StringOperator COUNT=new StringOperator("count");
	public static final StringOperator DELAY=new StringOperator("delay");
	public static final StringOperator TIMEUNIT=new StringOperator("timeUnit");
	
	private java.util.concurrent.ScheduledExecutorService eService;

	@Override
	protected boolean checkOperator(int index, Operator<?> operator) {
		if(index==0&&!ClosureOperator.class.isAssignableFrom(operator.getClass())) return false;
		if(index==1&&!MapOperator.class.isAssignableFrom(operator.getClass())) return false;
		return super.checkOperator(index, operator);
	}

	@SuppressWarnings("unchecked")
	@Override
	public <C,R,P> Operator<?> doInvoke(final CalContext<C,R,P> context,Operator<?>... operator) throws InvokeException{
		final ClosureOperator call=ClosureOperator.class.cast(operator[0]);
		final MapOperator<Operator<?>> map=MapOperator.class.cast(operator[1]);
		int count=1;
		if(map.value().get(COUNT)!=null){
			count=NumberOperator.class.cast(map.value().get(COUNT)).value().intValue();
		}
		int pool=1;
		if(map.value().get(POOL)!=null){
			pool=NumberOperator.class.cast(map.value().get(POOL)).value().intValue();
		}
		eService=java.util.concurrent.Executors.newScheduledThreadPool(pool);
		for(int i=0;i<count;i++){
			java.security.AccessController.doPrivileged(new PrivilegedAction<Operator<?>>(){
				@Override
				public Operator<?> run() {
					long delay=0;
					if(map.value().get(DELAY)!=null){
						delay=NumberOperator.class.cast(map.value().get(DELAY)).value().intValue();
					}
					TimeUnit timeUnit=TimeUnit.SECONDS;
					if(map.value().get(TIMEUNIT)!=null){
						timeUnit=TimeUnit.valueOf(StringOperator.class.cast(map.value().get(TIMEUNIT)).value());	
					}
					try {
						return eService.schedule(new Callable<Operator<?>>(){
							@Override
							public Operator<?> call() throws Exception {
								return call.execute(context.clone());
							}
							
						},delay,timeUnit).get();
					} catch (Exception e) {
						logger.error(e);
						return Operator.nullInstance;
					}
					
				}
			});	
		}
		return Operator.nullInstance;
	}
	

	
	
	@Override
	protected int minOperatorCount() {
		return 1;
	}
	
}
